﻿using Avro;
using Avro.Generic;
using Confluent.Kafka;
using Confluent.SchemaRegistry;
using Confluent.SchemaRegistry.Serdes;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace ProducerTest1
{
    public class AvroGenericProducer : IPush
    {
        private readonly string _schemaUri;
        private readonly string _topic;
        private readonly RecordSchema _rs;
        public AvroGenericProducer(string schemaUri, string topic)
        {
            _topic = topic;
            _schemaUri = schemaUri;
            _rs = (RecordSchema)RecordSchema.Parse(@"{
	                ""namespace"": ""kafkalearn.ProducerTest1.AvroSpecific"",
	                ""type"": ""record"",
	                ""name"": ""User"",
	                ""fields"": [
		                { ""name"": ""name"", ""type"": ""string"" },
		                { ""name"": ""favorite_number"", ""type"": [""int"",""null""] },
		                { ""name"": ""favorite_color"", ""type"": [""string"",""null""] },
		                { ""name"": ""hourly_rate"", ""type"": {
			                ""type"": ""bytes"",
			                ""logicalType"": ""decimal"",
			                ""precision"": 4,
			                ""scale"": 2
		                } }
	                ]
                }");
        }

        public void Invoke(ProducerConfig config)
        {
            throw new NotImplementedException();
        }

        public async Task InvokeAsync(ProducerConfig config)
        {
            SchemaRegistryConfig schemaRegistryConfig = new SchemaRegistryConfig
            {
                Url = _schemaUri
            };
            
            CancellationTokenSource cts = new CancellationTokenSource();
            using (var schemaRegistry = new CachedSchemaRegistryClient(schemaRegistryConfig))
            using (var producer = new ProducerBuilder<string, GenericRecord>(config)
            .SetKeySerializer(new AvroSerializer<string>(schemaRegistry))
            .SetValueSerializer(new AvroSerializer<GenericRecord>(schemaRegistry))
            .Build())
            {
                Console.WriteLine($"{producer.Name} producing on {_topic}. Enter user names, q to exit.");
                int i = 0;
                string text;
                while ((text = Console.ReadLine()) != "q")
                {
                    GenericRecord record = new GenericRecord(_rs);
                    record.Add("name", text);
                    record.Add("favorite_number", i++);
                    record.Add("favorite_color", "blue");
                    record.Add("hourly_rate", new AvroDecimal(67.99));
                    try
                    {
                        var dr = await producer.ProduceAsync(_topic, new Message<string, GenericRecord> { Key = text, Value = record });
                        Console.WriteLine($"produced to: {dr.TopicPartitionOffset}");
                    }
                    catch (ProduceException<string, GenericRecord> ex)
                    {
                        Console.WriteLine($"error producing message: {ex}");
                    }
                }
            }
            cts.Cancel();
        }
    }
}
